package com.only4play.importapt.config;

import com.only4play.idgenerator.provider.IdGenerator;
import com.only4play.idgenerator.provider.IdGeneratorOptions;
import com.only4play.idgenerator.provider.IdGeneratorProvider;
import com.only4play.idgenerator.strategy.IdGenerateStrategy;
import com.only4play.importapt.model.WrapTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * @author liyuncong
 * @version 1.0
 * @file ImportAptConfiguration
 * @brief ImportAptConfiguration
 * @details ImportAptConfiguration
 * @date 2024-01-19
 *
 * Edit History
 * ----------------------------------------------------------------------------
 * DATE                     NAME               DESCRIPTION
 * 2024-01-19               liyuncong          Created
 */

@Slf4j
@Configuration
public class ImportAptConfiguration {

    // 模拟缓存/DB存储拒绝任务数据
    private final CopyOnWriteArrayList<WrapTask> cacheTask = new CopyOnWriteArrayList<>();

    @Bean
    public IdGeneratorProvider idGeneratorProvider() {
        IdGeneratorOptions options = new IdGeneratorOptions();
        options.setGenerateStrategy(IdGenerateStrategy.UUID);
        return new IdGeneratorProvider(options);
    }

    @Bean
    public IdGenerator idGenerator() {
        return idGeneratorProvider().instance();
    }

    @Bean
    public ExecutorService executorService() {
        // 为滑动时间窗口准备一个队列存放元素的集合，用queue实现滑动时间窗口
        Queue<Integer> windowQueue = new LinkedList<>();

        // 获取可用线程
        int processors = Runtime.getRuntime().availableProcessors();
        log.info("Available processor is {} ", processors);
        final int minThreads = 2;
        final int queueSize = 5000;
        if (processors < minThreads) {
            processors = minThreads;
        }

//        return new ThreadPoolExecutor(processors, processors*2, 2, TimeUnit.HOURS, new LinkedBlockingQueue<>());

        return new ThreadPoolExecutor(processors, processors * 2, 2,
            TimeUnit.HOURS, new LinkedBlockingDeque<>(queueSize),
            // 自定义线程拒绝策略
            new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    WrapTask task = (WrapTask) r;

                    // 任务拒绝时的处理逻辑
//                    System.out.println("存入DB或者发送到MQ等待重试，此时retry_epoch为0！");
                    task.setRetryEpoch(0);
                    cacheTask.add(task);
                    log.info("cacheTask size={}", cacheTask.size());

                    /**
                     * 为什么滑动时间窗口写在此处，有什么作用，具体是为了什么：
                     * 1. 如果有100万条数据，被拒绝了100条，完全可以直接走重试机制，因为拒绝的条数占比非常小；
                     * 2. 但是100万条数据，被拒绝了10万条甚至更多，那么说明线程池参数定义有问题，或者是其他性能的问题，此时不建议走重试机制，
                     * 而是直接走报警进行人工干预；
                     * 3. 滑动时间窗口就是为此而诞生的：一段时间内，如果有n条以内的数据被拒绝，那么我们可以直接进行重试，超过n条数据就发送告警，
                     * 进行人工干预，用这种方式来避免雪上加霜的重试以及没有必要的告警；
                     */

                    // 滑动时间窗口判断是否需要进行告警，以便进行人工干预
                    // 设置时间窗口
                    int windowSize = 300;
                    // 获取当前时间戳
                    long currentTimestamp = System.currentTimeMillis();

                    // 如果队列不为空且最早入队的元素已经超过了指定的时间窗口大小（单位为秒），则发送警告，人工干预，并移除最老的元素
                    if (!windowQueue.isEmpty() && (currentTimestamp - windowQueue.peek()) > windowSize * 1000) {
                        // 如果单位时间内，拒绝条数超过设定的阈值，则发送告警，进行人工干预
                        if (windowQueue.size() > queueSize * 0.1) {
                            // 发送告警，进行人工干预
                            log.warn("-- 位时间内，拒绝条数超过设定的阈值，发送告警 --");
                        }

                        // 删除队列头
                        windowQueue.poll();
                    }

                    // 将新元素加入到队列尾部
                    windowQueue.offer((int) (System.currentTimeMillis()));
                }
            }) {

            /**
             * 在Worker线程执行任务之前会调用的方法
             * @param t the thread that will run task {@code r}
             * @param r the task that will be executed
             */
            @Override
            protected void beforeExecute(Thread t, Runnable r) {
                // 执行任务前
                this.getCorePoolSize(); // ：获取核心线程数。
                this.getMaximumPoolSize(); // ：获取最大线程数。
                this.getQueue(); // ：获取线程池中的阻塞队列，并通过阻塞队列中的方法获取队列长度、元素个数等。
                this.getPoolSize(); // ：获取线程池中的工作线程数（包括核心线程和非核心线程）。
                this.getActiveCount(); // ：获取活跃线程数，也就是正在执行任务的线程。
                this.getLargestPoolSize(); // ：获取线程池曾经到过的最大工作线程数。
                this.getTaskCount(); // ：获取历史已完成以及正在执行的总的任务数量。
                // 获取线程池的各项指标，将这些指标信息发送到mq中，消费端通过指标生成监控图，以此来了解线程池的执行情况以及它的负载问题
                // 还可以进行一些其他操作，比如一些日志记录，一些类的转换
            }

            /**
             * 在Worker线程执行任务之后会调用的方法
             * @param r the runnable that has completed
             * @param t the exception that caused termination, or null if
             * execution completed normally
             */
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                // 任务执行后
                // 进行重试的逻辑，会去看滑动窗口的queue是否有数据，如果有数据至少能证明有被拒绝的数据，查看当前线程池的queue里有多少没有处理的任务，
                // 用此来判断我们是否可以开始重试任务，并且在每次重试任务的时候，为重试纪元加一，超过重试的次数就进行报警，进行人工的干预
                // 线程池执行队列值小于设定值一半时，执行重试
                if (!windowQueue.isEmpty() && this.getQueue().size() < queueSize / 2) {
                    // 重试被拒绝的任务

                    // retryTasks模拟为从DB或MQ里读取的拒绝的任务列表, 此处取出一部分任务进行重试
                    int retrySize = queueSize - this.getQueue().size();
                    for (int i = 0; i < retrySize; i++) {
                        WrapTask myTask = cacheTask.get(i);

                        int retryEpoch = myTask.getRetryEpoch();
                        // 比如说当前任务被拒绝3次
                        if (retryEpoch > 3) {
                            // 发送告警
                            log.warn("-- 当前任务重试次数到达阈值，发送告警 --");
                            return;
                        }
                        myTask.setRetryEpoch(retryEpoch + 1);
                        this.execute(myTask);

                        // 删除缓存
                        cacheTask.remove(myTask);
                        log.info("删除缓存任务");
                    }
                } else {
                    cacheTask.forEach(myTask -> {
                        int retryEpoch = myTask.getRetryEpoch();
                        // 比如说当前任务被拒绝3次
                        if (retryEpoch > 3) {
                            // 发送告警
                            log.warn("-- 当前任务重试次数到达阈值，发送告警 --");
                            return;
                        }
                        myTask.setRetryEpoch(retryEpoch + 1);
                        this.execute(myTask);

                        // 删除缓存
                        cacheTask.remove(myTask);
                        log.info("删除缓存任务");
                    });
                }
            }

            /**
             * 当线程池从状态变更到TERMINATED状态之前调用的方法
             */
            @Override
            protected void terminated() {
                super.terminated();
            }
        };
    }
}
